#
# Translation from styx to op for ofs.
# Either use the Cache or Op to operate on remote files.
#
# semantics:
#	dir (+initial data) fetched/checked upon walk
#	directories fully read.
#	files read on demand (after data read on first get)
#	creates deferred until first write, but for "mkdirs".
#	writethrough:
#		data pushed for write at 0; and partially filled puts
#	delayed write:
#		data pushed with full puts.
#
# All files with open fids are alive in the cache.
#
# A coherency lag of N ms can be set. In this case,
# files checked out not more than Nms ago are considered
# up to date. For directories, this is tricky, because we must
# still read directories for which we got just metadata, and not data.
# in particular, directories invented during a walk.
#
# Each close in a file being written signals an end-of-put (clunk)
# in the server. However, this does not make any distinction between different
# client processes updating the same file in the server (which is reasonable, because
# in that case there would be races in any case).

implement Stop;
include "sys.m";
	sys: Sys;
	fprint, sprint, pctl, millisec, DMDIR, write, nulldir, tokenize, fildes,
	QTDIR, FD, Dir, Qid: import sys;
include "styx.m";
	styx: Styx;
	unpackdir: import styx;
include "styxservers.m";
	Navop: import Styxservers;
	Enotfound, Eexists: import Styxservers;
include "op.m";
	op: Op;
	Tmsg, Rmsg, OCREATE, MAXDATA, NOFD, OMORE, OSTAT, ODATA: import op;
include "opmux.m";
	opmux: Opmux;
	rpc: import opmux;
include "names.m";
	names: Names;
	cleanname: import names;
include "string.m";
	str: String;
	splitstrl: import str;
include "cache.m";
	cache: Cache;
	Creq, Crep, MAXNMSGS, MAXCACHED, cacherpc: import cache;
include "error.m";
	err: Error;
	kill, stderr, error, panic: import err;
include "stop.m";


# We keep up to MAXCACHED in f.data


init(cop: Op, cmux: Opmux, s: Styx, c: Cache): string
{
	sys = load Sys Sys->PATH;
	err = load Error Error->PATH;
	err->init();
	names = load Names Names->PATH;
	str = load String String->PATH;
	if (sys == nil || names == nil || str == nil)
		return "load failed in cache";
	styx = s;
	op = cop;
	opmux= cmux;
	cache = c;
	return nil;
}

term()
{
	kill(asyncpid, "kill");
}


#
# Main entry points
#
# These functions execute in the context of a client process. When they
# need to access the tree, they ask the central cache fs proc through
# a channel,  to prevent races when using the cache.
# Op is spoken but the functions themselves, using opmux. Otherwise
# this would lead to one request at a time.

dump()
{
	cacherpc(ref Creq.Dump(big 0, nil));
}

stat(qid: big): (ref Sys->Dir, string)
{
	r := cacherpc(ref Creq.Stat(qid, nil));
 	pick rr := r {
	Stat =>
		if (rr.err != nil)
			return (nil, rr.err);
		else
			return (rr.d, rr.err);
	}
	panic("statbug");
	return (nil, "statbug");
}

validate(qid: big, name: string): (ref Dir, string)
{
	r := cacherpc(ref Creq.Info(qid, nil));
	pick rr := r {
	Info =>
		f := rr.f;
		v := cacherpc(ref Creq.Valid(qid, f, name));
		pick vv := v {
		Valid =>
				if (vv.d == nil || f == nil){
					if (vv.err == nil)
						return(vv.d, vv.uerr);
				} else {
					if (vv.err == nil)
					if ((vv.d.qid.qtype&QTDIR) == 0 || f.dirread)
						return (vv.d, vv.uerr);
				}
		}
		path := opfname(rr.path, name);
		if (rr.err == nil && rr.dirty)
			return (rr.d, nil);
		(d, data, e) := get(path, OSTAT|ODATA, big 0);
		if (e == nil){
			v = cacherpc(ref Creq.Validate(qid, f, name, d, data));
			pick vv := v {
			Validate =>
				d = vv.d;
			}
			return (d, nil);
		} else {
			cacherpc(ref Creq.Invalidate(qid, f, name));
			return (nil, e);
		}
	}
	panic("validatebug");
	return (nil, "validatebug");
}

walk1(qid: big, name: string): (ref Dir, string)
{
	r := cacherpc(ref Creq.Walk1(qid, nil, name));
 	pick rr := r {
	Walk1 =>
		if (rr.err != nil)
			return (nil, rr.err);
		else
			return (rr.d, nil);
	}
	panic("walk1bug");
	return (nil, "walk1bug");
}

# cnt 0 means all of them.
readdir(qid: big,  off, cnt: int): (list of Sys->Dir, string)
{
	r := cacherpc(ref Creq.Readdir(qid, nil, off, cnt));
	pick rr := r {
	Readdir =>
		if (rr.err != nil)
			return (nil, rr.err);
		else
			return (rr.sons, rr.err);
	* =>
		panic("readdirbug");
		return (nil, "readdirbug");
	}
}

# Also used to truncate
wstat(qid: big, d: ref Sys->Dir): (ref Sys->Dir, string)
{
	r := cacherpc(ref Creq.Wstat(qid, nil, d, 1));
	pick rr := r {
	Wstat =>
		if (rr.err != nil)
			return (nil, rr.err);
		else
			return (rr.d, rr.err);
	}
	panic("wstatbug");
	return (nil, "wstatbug");
}

pread(qid: big, cnt: int, off: big): (array of byte, string)
{
	r := cacherpc(ref Creq.Pread(qid, nil, cnt, off));
  	pick rr := r {
	Pread =>
		if (rr.data != nil)
			return (rr.data, nil);
		if (rr.err != nil && rr.path != nil) {
			(nil, rr.data, rr.err) = get(rr.path, ODATA, off);
			if (rr.err == nil)
				cacherpc(ref Creq.Pwrite(qid, r.f, rr.data, off, 0));
			if (len rr.data < cnt)
				cnt = len rr.data;
			if (rr.err == nil)
				return (rr.data[0:cnt], nil);
		}
		return (nil, rr.err);
	}
	panic("preadbug");
	return (nil, "preadbug");
}

# Pwrite does write-through for first and last put, it does delayed
# write otherwise. The first write-through stores the Op fd in the
# cache, and further puts use that fd (ie., the server would keep
# the file "open" for all these puts).
# Upon a clunk (see :/^flush below), a final put is sent without OMORE set,
# to clear the Op fd (ie., the server would close it).

pwrite(qid: big, data: array of byte, off: big) : (int, string)
{
	r := cacherpc(ref Creq.Pwrite(qid, nil, data, off, 1));
  	pick rr := r {
	Pwrite =>
		if (r.err != nil)
			return(-1, r.err);
		f := cacherpc(ref Creq.Flush(qid, rr.f));
		pick ff := f{
		Flush =>
			if (ff.err != nil)
				return(-1, ff.err);
			# BUG: coherency lag, and hardwired iosize
			# also, this updates ff.opfd, and creates a race.
			# The time for a full rewrite of cache is approaching...
			err : string;
			nvers: int;
			if (len data == 8192 && !ff.created && off != big 0)
				put(rr.path, rr.f.opfd, ff.d, data, off, ff.created, OMORE, 1);
			else
				(err, nvers, rr.f.opfd) = put(rr.path, rr.f.opfd, ff.d, data, off, ff.created, OMORE, 0);
			if (err != nil)
				return(-1, err);
			nd := ref nulldir;
			nd.qid.vers = nvers;
			cacherpc(ref Creq.Wstat(qid, rr.f, nd, 0)); 
			return (len data, nil);
		* =>
			panic("pwrite bug");
		}
	* =>
		panic("pwritebug");
	}
	panic("pwritebug");
	return (-1, "pwritebug");
}

flush(qid: big) : string
{
	# BUG: Tclunk is too late in Inferno to report failures.
	# we must notify clients on writes.
	r := cacherpc(ref Creq.Flush(qid, nil));
  	pick rr := r {
	Flush =>
		if (rr.err == nil && (rr.d != nil || rr.created || rr.f.opfd != NOFD )){
			put(rr.path, rr.f.opfd, rr.d, nil, big 0, rr.created, 0, 1);
			rr.f.opfd = NOFD;
		}
		return nil;
	}
	panic("flushbug");
	return "flushbug";
}

create(qid: big, d: ref Sys->Dir): (ref Sys->Dir, string)
{
	r := cacherpc(ref Creq.Create(qid, nil, d.name, d, array[0] of byte));
	pick rr := r {
	Create =>
		if (rr.err == nil && (rr.d.qid.qtype&QTDIR))
			rr.err = flush(rr.d.qid.path);
		return (rr.d, rr.err);
	}
	panic("createbug");
	return (nil, "createbug");
}

remove(qid: big) : string
{
	r := cacherpc(ref Creq.Remove(qid, nil));
	pick rr := r {
	Remove =>
		oprc := rpc(ref Tmsg.Remove(0, rr.path));
		opr := <-oprc;
		pick oprr := opr {
		Remove =>
			return nil;
		Error =>
			return oprr.ename;
		}
	}
	panic("removebug");
	return "removebug";
}

#
# Helper routines, speaking Op
#

opfname(d, n: string) : string
{
	path: string;
	if (n == "" || n == "/")
		path = d;
	else if (d == "/")
		path = n;
	else
		path = d + "/" + n;
	path = cleanname(path);

	return path;
}

delayed(rc: chan of ref Rmsg)
{
	rm := <-rc;
	pick rr := rm {
	Readerror =>
		fprint(stderr, "*** ofs: delayed write (read)error ***\n");
	Error =>
		fprint(stderr, "*** ofs: delayed write error ***\n");
	}
}

asyncc : chan of chan of ref Rmsg;
asyncpid:= -1;
asyncproc()
{
	asyncpid = pctl(0, nil);
	for(;;){
		c :=<-asyncc;
		<- c;
		c = nil;
	}
}

# BUG: Too many args; put does it all.
put(path: string, opfd: int, d: ref Dir, data: array of byte, off: big, mkit: int, more: int, async: int) : (string, int, int)
{
	if (d == nil && data == nil && opfd == NOFD)
		panic("bug: no changes for put");
	flag := 0;
	if (d != nil)
		flag |= OSTAT;
	if (data != nil && (!mkit || (d.qid.qtype&QTDIR) == 0))
		flag |= ODATA;
	if (mkit){
		opfd = NOFD;
		flag |= OCREATE;
	}
	if (more)
		flag |= OMORE;
	if (data == nil)
		data = array[0] of byte;
	tot := 0;
	rfd := NOFD;
	for(;;){
		nw := len data[tot:];
		if (nw > MAXDATA)
			nw = MAXDATA;
		if (d == nil)
			d = ref nulldir;
		rc := rpc(ref Tmsg.Put(0, path, opfd, flag, *d, off, data[tot:tot+nw]));
		vers := 0;
		if (!async) {
			r := <- rc;
			pick rr := r {
			Put =>
				if (rr.count != nw)
					return ("short write", 0, NOFD);
				vers = rr.qid.vers;
				rfd = rr.fd;
			Error =>
				return (rr.ename, 0, NOFD);
			* =>
				panic("putbug");
			}
		}
		tot += nw;
		if (tot == len data)
			return (nil, vers, rfd);
		flag &= ~(OSTAT|OCREATE);
	}
	panic("putbug");
	return ("putbug", 0, NOFD);
}


# initial get for a file; used by walk.
get(path: string, mode: int, off: big): (ref Dir, array of byte, string)
{
	data := array[0] of byte;
	d : ref Dir;

	nm := 1;
	while(nm < MAXNMSGS && off + big (MAXDATA*nm) < big MAXCACHED)
		nm++;

	req := ref Tmsg.Get(0, path, mode, nm, off, MAXDATA);
	repc := rpc(req);
	nmsgs := nm;
	for (;;){
		rep := <- repc;
		pick r := rep {
		Error =>
			return (nil, nil, r.ename);
		Get =>
			if ((r.mode&OSTAT) != 0){
				d = ref r.stat;
				if ((d.qid.qtype&QTDIR) != 0)
					nmsgs = 0;
			}
			if ((r.mode &ODATA) != 0){
				ndata := array[len data + len r.data] of byte;
				ndata[0:] = data;
				ndata[len data:] = r.data;
				data = ndata;
			}
			if (--nmsgs == 0 || (r.mode&OMORE) == 0)	# last message
				return(d, data, nil);
		* =>
			panic("get0bug");
		}
	}
	panic("get0bug");
	return (nil, nil, "bug");
}

